package demoflink.watermark;

import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkOutput;

public class MyPeriodWatermarkStrategy<T> implements WatermarkGenerator<T> {
    private long delayTs;
    private long maxTs;

    public MyPeriodWatermarkStrategy(long delayTs) {
        this.delayTs = delayTs;
        this.maxTs = Long.MIN_VALUE+delayTs+1;
    }

    /**
     * 每条数据来斗湖调用一次
     * @param event
     * @param eventTimestamp
     * @param watermarkOutput
     */
    @Override
    public void onEvent(T event, long eventTimestamp, WatermarkOutput watermarkOutput) {


        maxTs = Math.max(maxTs,eventTimestamp);
        System.out.println("获取目前为止的最大时间戳"+maxTs);

    }

    /**
     * 周期性调用
     * @param watermarkOutput
     */
    @Override
    public void onPeriodicEmit(WatermarkOutput watermarkOutput) {
        watermarkOutput.emitWatermark(new Watermark(maxTs-delayTs-1));
        System.out.println("调用onPeriodEmit 方法,生成watermark="+(maxTs-delayTs-1));

    }
}
